package com.hy.study.mapper;

import com.hy.study.bean.JoinTableBean;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * @author huyong
 * @date 2024 年 02 月 06 日 14:10: 38
 */
public class JoinTableMapper extends Mapper<LongWritable, Text, Text, JoinTableBean> {
    private String fileName;
    private final JoinTableBean v = new JoinTableBean();
    private final Text k = new Text();


    @Override
    protected void setup(Mapper<LongWritable, Text, Text, JoinTableBean>.Context context) throws IOException, InterruptedException {
        // 获取文件的名称 通过切片获取
        FileSplit fileSplit = (FileSplit) context.getInputSplit();
        fileName = fileSplit.getPath().getName();
    }

    /**
     * 1001	01	1
     * 1002	02	2
     * 1003	03	3
     * 1004	01	4
     * 1005	02	5
     * 1006	03	6
     *
     * 01	小米
     * 02	华为
     * 03	格力
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, JoinTableBean>.Context context) throws IOException, InterruptedException {
        // 封装表的数据到bean
        String line = value.toString();
        String[] split = line.split("\t");
        if (fileName.startsWith("order")) {// 订单
            v.setId(split[0]);
            v.setPid(split[1]);
            v.setAmount(Integer.parseInt(split[2]));
            v.setPname("");
            v.setFlag("order");
            k.set(split[1]);
        } else {// 产品表
            v.setId("");
            v.setPid(split[0]);
            v.setAmount(0);
            v.setPname(split[1]);
            v.setFlag("pd");
            k.set(split[0]);
        }
        context.write(k, v);
    }
}
